package batchMessage;

import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.List;
import java.util.function.Consumer;

import com.alibaba.rocketmq.client.exception.MQBrokerException;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.MQProducer;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.remoting.exception.RemotingException;

public class BatchProducer {
    public static void main(String[] args) throws MQClientException, UnsupportedEncodingException, RemotingException,
            MQBrokerException, InterruptedException {
        final MQProducer producer = new DefaultMQProducer();

        String topic = "BatchTest";
        List<Message> messages = new ArrayList<Message>();
        messages.add(new Message(topic, "TagA", "orderID001", "hello world 0".getBytes()));
        messages.add(new Message(topic, "TagA", "orderID002", "hello world 1".getBytes()));
        messages.add(new Message(topic, "TagA", "orderID003", "hello world 2".getBytes()));

        //then you could split the large list into small ones:
        ListSplitter splitter = new ListSplitter(messages);
        while (splitter.hasNext()) {

            List<Message> listItem = splitter.next();
            listItem.forEach(consumerWrapper(message -> {
                try {
                    producer.send(message);
                } catch (MQClientException e) {
                    e.printStackTrace();
                } catch (RemotingException e) {
                    e.printStackTrace();
                } catch (MQBrokerException e) {
                    e.printStackTrace();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }, Exception.class));


        }

        producer.shutdown();
    }

    static Consumer<Message> lambdaWrapper(Consumer<Message> consumer) {
        return message -> {
            try {
                consumer.accept(message);
            } catch (Exception e) {

            }

        };
    }

    static <T, E extends Exception> Consumer<T> consumerWrapper(Consumer<T> consumer, Class<E> clazz) {
        return (T i) -> {
            try {
                consumer.accept(i);
            } catch (Exception e) {

                E ex = clazz.cast(e);
            }
        };
    }
}
